
package com.naza.rpc.core;

import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.naza.rpc.circuitBreaker.CircuitBreaker;
import com.naza.rpc.exception.InvokeModuleException;
import com.naza.rpc.exception.TimeoutException;
import com.naza.rpc.exception.RejectResponeException;
import com.naza.rpc.parallel.RpcThreadPool;
import com.naza.rpc.async.RpcCallback;
import com.naza.rpc.model.RpcRequest;
import com.naza.rpc.model.RpcResponse;

/**
 * Rpc客户端消息回调
 * 
 * @author yl
 */
public class MessageCallBack {

    private RpcRequest request;

    private RpcResponse response;

    private RpcCallback rpcCallback;

    private CircuitBreaker circuitBreaker;

    private Lock lock = new ReentrantLock();

    private Condition finish = lock.newCondition();

    private ThreadPoolExecutor executor = (ThreadPoolExecutor) RpcThreadPool.getExecutor(RpcSystemConfig.SYSTEM_PROPERTY_THREADPOOL_THREAD_NUMS, RpcSystemConfig.SYSTEM_PROPERTY_THREADPOOL_QUEUE_NUMS);

    public MessageCallBack(RpcRequest request,CircuitBreaker circuitBreaker) {
        this.request = request;
        this.circuitBreaker = circuitBreaker;
    }

    /**
     * 设置异步回调
     * @return
     */
    public void setRpcCallback(RpcCallback callback) {
        this.rpcCallback = callback;
    }

    /**
     *  同步调用,异步转同步等待
     * @return
     */
    public Object start() {
        try {
            lock.lock();
            await();
            if (this.response != null) {
                boolean isInvokeSucc = getInvokeResult();
                if (isInvokeSucc) {
                    // invoke success
                    if (this.response.getError().isEmpty()) {
                        return this.response.getResult();
                    } else {
                        throw new InvokeModuleException(this.response.getError());
                    }
                } else {
                    //invoke fail
                    throw new RejectResponeException(RpcSystemConfig.FILTER_RESPONSE_MSG);
                }
            } else {
                return null;
            }
        } finally {
            lock.unlock();
        }
    }
    /**
     * 收到服务端返回的响应，唤醒回调上等待的线程(finish)
     * @param reponse
     */
    public void over(RpcResponse reponse) {
        try {
            lock.lock();
            finish.signal();
            this.response = reponse;
            if(null != rpcCallback){
                notifyCallBack();
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 唤醒回调
     */
    private void notifyCallBack(){
        Runnable task = new Runnable() {
            @Override
            public void run() {
                if (response != null) {
                    boolean isInvokeSucc = getInvokeResult();
                    if (isInvokeSucc) {
                        // invoke success
                        if (response.getError().isEmpty()) {
                            rpcCallback.onSuccess(response.getResult());
                        } else {
                            rpcCallback.onError(new InvokeModuleException(response.getError()));
                        }
                    } else {
                        rpcCallback.onError(new InvokeModuleException(response.getError()));
                    }
                }
            }
        };
        executor.execute(task);
    }

    /**
     * 客户端向服务端发送请求之后，客户端回调线程等待的超时时间，超过时间抛异常
     */
    private void await() {
        boolean timeout = false;
        try {
            // 设定一下超时时间,rpc服务器太久没有相应的话，就默认返回空
            timeout = finish.await(RpcSystemConfig.SYSTEM_PROPERTY_MESSAGE_CALLBACK_TIMEOUT,
                    TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (!timeout) {
            throw new TimeoutException(RpcSystemConfig.TIMEOUT_RESPONSE_MSG);
        }
    }

    /**
     * 服务端返回正常结果
     * @return
     */
    private boolean getInvokeResult() {
        boolean invokeResult =  (!this.response.getError().equals(RpcSystemConfig.FILTER_RESPONSE_MSG)
                && (!this.response.isReturnNotNull()
                        || (this.response.isReturnNotNull() && this.response.getResult() != null)));
        if(invokeResult){
            //调用成功
            circuitBreaker.recordSuccess();
        }else {
            //调用失败
            circuitBreaker.recordFailure();
        }
        return invokeResult;
    }
}
